package mapreduce.phase2;

/**
 * Get the global picture of the shared points
 * Reduce: apply union find
 */


import java.io.IOException;
import java.util.Iterator;
import java.util.TreeSet;
import mapreduce.Point;
import mapreduce.PointComparator;
import mapreduce.UnionFind;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

@SuppressWarnings("deprecation")
public class Reduce2 extends MapReduceBase implements
Reducer<IntWritable, Text, IntWritable, Text> {

	public void reduce(IntWritable key, Iterator<Text> values,
			OutputCollector<IntWritable, Text> output, Reporter reporter)
	throws IOException {

		UnionFind uf = new UnionFind();
		TreeSet<Point> points = new TreeSet<Point>(new PointComparator());
		while (values.hasNext()) {
			String s[] = values.next().toString().split(" ");
			Point p1 = Point.fromString(s[0]);
			uf.makeSet(p1);
			Point p2 = Point.fromString(s[1]);
			uf.makeSet(p2);
			uf.union(p1, p2);
			points.add(p1);
		}

		for (Point p : points) {
			Text t2 = new Text(p + " " + uf.find(p));
			output.collect(key, t2);
		}
	}
}

